package com.tt.job.executor;

import com.tt.job.api.JobContext;
import com.tt.job.api.JobResult;
import com.tt.job.api.type.dataflow.DataFlowJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;

/**
 * 针对数据流任务的处理执行
 */
public class DataFlowJobMessageExecutor {
    private final static Logger logger = LoggerFactory.getLogger(DataFlowJobMessageExecutor.class);
    private final ScheduledThreadPoolExecutor executorService;
    private final DataFlowJob job;
    int failedCount = 0;
    int skipCount = 0;
    int successCount = 0;

    public DataFlowJobMessageExecutor(ScheduledThreadPoolExecutor executorService, DataFlowJob job) {
        this.executorService = executorService;
        this.job = job;
    }

    /**
     * 处理数据流任务
     *
     * @param jobContext
     * @return
     */
    public JobResult execute(JobContext jobContext) {
        logger.debug("开始进行dataflow任务处理, job:{}, schedule:{}", jobContext.getJobClass(), jobContext.getSchedule());
        if (job.isConcurrent()) {
            return concurrentExecute(jobContext);
        } else {
            return orderExecute(jobContext);
        }
    }

    /**
     * 并发的方式进行处理
     *
     * @param jobContext
     * @return
     */
    private JobResult concurrentExecute(JobContext jobContext) {
        List data = job.fetchData(jobContext);
        Deque<Future<JobResult>> futures = new ArrayDeque<>();

        while (data != null) {
            final List innerData = data;
            Future<JobResult> resultFuture = executorService.submit(new Callable<JobResult>() {
                @Override
                public JobResult call() throws Exception {
                    return job.processData(jobContext, innerData);
                }
            });
            futures.addFirst(resultFuture);

            clearDoneFuture(futures);
            data = job.fetchData(jobContext);
        }
        //等待所有任务处理完成
        while (futures.size() > 0) {
            clearDoneFuture(futures);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {

            }
        }
        //所有任务均处理完成，只要有任务处理完成就认为成功
        logger.debug("dataflow任务并发处理完成,job:{}, success:{}, failed:{}, skip:{}", jobContext.getJobClass(), successCount, failedCount, skipCount);
        return successCount > 0 ? JobResult.SUCCESS : JobResult.FAILED;
    }

    /**
     * 顺序方式处理
     *
     * @param jobContext
     * @return
     */
    private JobResult orderExecute(JobContext jobContext) {
        List data = job.fetchData(jobContext);
        while (data != null) {
            JobResult result = job.processData(jobContext, data);
            //对本次的处理结果进行统计
            if (result == JobResult.SUCCESS) {
                successCount++;
            } else if (result == JobResult.SKIP) {
                skipCount++;
            } else {
                failedCount++;
            }
            data = job.fetchData(jobContext);
        }
        logger.debug("dataflow任务顺序处理完成,job:{}, success:{}, failed:{}, skip:{}", jobContext.getJobClass(), successCount, failedCount, skipCount);
        //所有任务均处理完成，只要有任务处理完成就认为成功
        return successCount > 0 ? JobResult.SUCCESS : JobResult.FAILED;
    }

    /**
     * 将队列中的完成任务信息移除，并统计
     *
     * @param futures
     */
    private void clearDoneFuture(Deque<Future<JobResult>> futures) {
        Future<JobResult> resultFuture;
        resultFuture = futures.peekLast();
        while (resultFuture != null && resultFuture.isDone()) {
            successCount++;
            JobResult result = null;
            try {
                result = futures.removeLast().get();
                if (result == JobResult.SUCCESS) {
                    successCount++;
                } else if (result == JobResult.SKIP) {
                    skipCount++;
                } else {
                    failedCount++;
                }
            } catch (Exception e) {

            }
            resultFuture = futures.peekLast();
        }
    }
}
